package com.bigdata.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * mapPartitions 遍历的是每个分区中的数据，一个个分区的遍历
 * 每个分组触发一次函数的执行, input Iterator; output Iterator
 * 相对于map 一条条处理数据，性能比较高。
 */
object Demo17_mapPartitions {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("mapPartitions")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val infos = sc.parallelize(List[String]("a", "b", "c", "d", "e", "f", "g"), 2)

    /*
    mapPartitions 每次处理的是一个分区的数据,也就是一批数据
    对这一批数据加了一个后缀，并返回
    得到的结果RDD类型和迭代器的类型一致，
    结果RDD里面的值就是每一条映射的结果
    返回集合的迭代器, 集合的泛型就是RDD的泛型

     */
    val result: RDD[String] = infos.mapPartitions(iter => {
      println("process partition data before")
      val arr = new ArrayBuffer[String]()
      iter.foreach(str => {
        println(str)
        println("process data.....")
        arr += str + "-bigdata"
      })
      println("process partition data after")

      arr.iterator
    })

    Thread.sleep(2000)

    // count是一个action算子，能够触发job
    result.foreach(println)

    println("---------------")

  }
}
